package net.bwie.realtime.vehicle.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.common.utils.JdbcUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.vehicle.bean.BatteryHealthData;
import net.bwie.realtime.vehicle.function.BatteryHealthWindowFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

/**
 * 平均电池健康度
 */
public class VehicleWindowBatteryHealthJob {
    public static void main(String[] args) throws Exception {
        // 获取流执行环境并设置并行度为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 从Kafka中消费电池数据
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle_battery_data");
        // 处理电池健康数据
        BatteryHealth(kafkaStream);
        env.execute("vehicleWindowBatteryHealth");
    }

    // 处理电池健康数据
    private static void BatteryHealth(DataStream<String> kafkaStream) {
        // 将Kafka中的数据转换为BatteryHealthData对象
        SingleOutputStreamOperator<BatteryHealthData> mapStream = kafkaStream.map(o -> JSON.parseObject(o, BatteryHealthData.class));
        // 为数据分配时间戳和水位线
        SingleOutputStreamOperator<BatteryHealthData> batteryHealthDataSingleOutputStreamOperator = mapStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<BatteryHealthData>forBoundedOutOfOrderness(Duration.ofSeconds(30)).withTimestampAssigner(
                        new SerializableTimestampAssigner<BatteryHealthData>() {
                            @Override
                            public long extractTimestamp(BatteryHealthData element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }
                )
        );
        // 按照时间窗口进行分组
        WindowedStream<BatteryHealthData, String, TimeWindow> window = batteryHealthDataSingleOutputStreamOperator.keyBy(o -> "all")
                .window(TumblingEventTimeWindows.of(Time.minutes(5)));
        // 对窗口中的数据应用BatteryHealthWindowFunction进行处理
        SingleOutputStreamOperator<String> apply = window.apply(new BatteryHealthWindowFunction());
        // 将处理结果写入ClickHouse数据库
        JdbcUtil.sinkToClickhouseUpsert(apply,
                "INSERT INTO jtp_vehicle_log.vehicle_dws_battery_Health(\n" +
                        "startTime, endTime, vehicleCount, sumTemp, avgTemp, ts)\n" +
                        "VALUES(?,?,?,?,?,?)");
    }
}
